package com.alan.weather.service.weather;

import com.alan.weather.entity.weather.WeatherData_PredictOneDay;
import com.alan.weather.entity.weather.WeatherData_PredictOneWeek;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;

public class Step3_6 {
    static WeatherData_PredictOneWeek result = new WeatherData_PredictOneWeek();
    private static class Step3_6Mappper extends Mapper<LongWritable, Text, Text, WeatherWritable> {
        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, WeatherWritable>.Context context) throws IOException, InterruptedException {
            String line = value.toString();
            // 数据验证
            if (StringUtils.isBlank(line)){
                return;
            }
            // 防止数据拆分
            String[] items = line.split("\t");
            if (items == null || items.length != 5){
                return;
            }
            // 获取查询日期
            String date = context.getConfiguration().get("date");

            int week = 7;
            SimpleDateFormat ft = new SimpleDateFormat("dd/MM/yyyy");
            Date time = null;
            try {
                time = ft.parse(date);
            } catch (ParseException e) {
                throw new RuntimeException(e);
            }
            Calendar calendar = new GregorianCalendar();
            String[] day_months = new String[7];
            while (week-- != 0) {
                calendar.setTime(time);
                date = ft.format(time);
                day_months[week] = date.substring(0, date.lastIndexOf("/"));
                calendar.add(Calendar.DATE,1);
                time = calendar.getTime();
            }

            // 过滤日期
            if ((items[0].indexOf(day_months[0])
                   + items[0].indexOf(day_months[1])
                    + items[0].indexOf(day_months[2])
                    + items[0].indexOf(day_months[3])
                    + items[0].indexOf(day_months[4])
                    + items[0].indexOf(day_months[5])
                    + items[0].indexOf(day_months[6]))< 0){
                return;
            }
            String code = items[0].split("_")[0];
            String day_month = items[0].split("_")[1].substring(0,5);
            // 实例化对象
            WeatherWritable w = new WeatherWritable(code + "_" + day_month,
                    Double.parseDouble(items[1]),
                    Double.parseDouble(items[2]),
                    Double.parseDouble(items[3]),
                    Double.parseDouble(items[4]));
            // 输出:<8377_31/12, w>
            context.write(new Text(code + "\t" + day_month), w);
        }
    }

    private static class Step3_6Reducer extends Reducer<Text, WeatherWritable, WeatherWritable, NullWritable> {
        @Override
        protected void reduce(Text key, Iterable<WeatherWritable> values, Reducer<Text, WeatherWritable, WeatherWritable, NullWritable>.Context context) throws IOException, InterruptedException {
            WeatherWritable w = new WeatherWritable(key.toString(), 0.0, 0.0, 0.0, 0.0);
            int flag = 0;
            for (WeatherWritable v : values){
                if (flag == 0){
                    w.setPrecipitation(v.getPrecipitation());
                    w.setMaxTemperature(v.getMaxTemperature());
                    w.setMinTemperature(v.getMinTemperature());
                    w.setAvgTemperature(v.getAvgTemperature());
                    flag++;
                }
                w.setPrecipitation((w.getPrecipitation() + v.getPrecipitation()) / 2);
                w.setMaxTemperature((w.getMaxTemperature() + v.getMaxTemperature()) / 2);
                w.setMinTemperature((w.getMinTemperature() + v.getMinTemperature()) / 2);
                w.setAvgTemperature((w.getAvgTemperature() + v.getAvgTemperature()) / 2);
            }
            // 输出
            context.write(w, NullWritable.get());
            WeatherData_PredictOneDay temp = new WeatherData_PredictOneDay();
            temp.date = w.getCode_date().split("\t")[1];
            temp.averageTemperature = w.getAvgTemperature().toString().substring(0, 5);
            temp.highestTemperature = w.getMaxTemperature().toString().substring(0, 5);
            temp.lowestTemperature = w.getMinTemperature().toString().substring(0, 5);
            temp.precipitation = w.getPrecipitation().toString().substring(0, 5);
            result.week.add(temp);
        }
    }

    public static WeatherData_PredictOneWeek run(String input, String output, String date) {
        try {
            result.week.clear();
            // 定义输出路径
            Path outputPath = new Path(output);
            // 判断输出目录是否存在，存在则删除之
            HadoopUtils.delete(HadoopUtils.getFileSystem(), outputPath, true);
            // 传递日期:通过配置对象将参数设置为全局
            HadoopUtils.getConf().set("date",date);
            // 定义Job任务
            Job job = Job.getInstance(HadoopUtils.getConf(), "step3_6");
            //设置Jar包
            job.setJarByClass(Step1.class);
            // 设置输入
            job.setInputFormatClass(TextInputFormat.class);
            FileInputFormat.setInputPaths(job, input);
            // 设置Mapper
            job.setMapperClass(Step3_6Mappper.class);
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(WeatherWritable.class);
            // 设置Reducer
            job.setReducerClass(Step3_6Reducer.class);
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(NullWritable.class);
            // 设置输出
            job.setOutputFormatClass(TextOutputFormat.class);
            FileOutputFormat.setOutputPath(job, outputPath);
            // 运行
            boolean flag = job.waitForCompletion(true);
            // 提示
            if (flag){
                System.out.println("Step3_6:预测指定日期及其未来7天（含指定那一天）的天气~~~");
                // 显示数据
                return result;
            }
        }
        catch (Exception e){
            e.printStackTrace();
        }
        return null;
    }
}
